package com.atguigu.flink.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * Created by Smexy on 2023/3/1
 *
 *  StateBackend； 是一个可拔插的组件。
 *                  负责读，写状态。
 *                  负责把状态持久化到一个设备
 *                      默认 HashMapStateBackend。 把定义的状态存储在TM的队内存中，用HashMap结构存储。 生产不可用。 有限的。
 *                                                  优势：快。 10倍
 *
 *                      RocksDBstateBackend: 内嵌式数据库。 类似 HBase。
 *                                              有列族，还有读写缓存。写缓存还会在快满的时候，刷写到磁盘。存储空间是无限的。
 *                                              数据以byte[]形式存储。 读写状态时，有序列化和发序列化的开销。
 *
 *                                              不需要安装，把进程启动起来，就自带。
 *                                                  优势： 大
 *
 *
 *  checkpoint:   负责定义状态持久化到哪里?
 *                  默认存储在 JobManagerCheckpointStorage。不可靠，生成不可用。最好是持久化！
 *                  可以设置: FileSystemCheckpointStorage
 */
public class Demo10_StateBackend
{

    public static void main(String[] args) {
/*
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);*/
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //env.setParallelism(2);
        env.setStateBackend(new EmbeddedRocksDBStateBackend());

        env.enableCheckpointing(2000);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        //持久化的存储，不用担心 JM崩溃造成状态丢失
        //checkpointConfig.setCheckpointStorage("file:///d:/tmp");
        checkpointConfig.setCheckpointStorage("hdfs://hadoop102:9820/ck");

        env
            .socketTextStream("hadoop103", 8888)
            .map(new MyMapFunction())
            .addSink(new SinkFunction<String>()
            {
                //如果数据中包含了 x，手动模拟异常
                @Override
                public void invoke(String value, SinkFunction.Context context) throws Exception {

                    if (value.contains("x")){
                        throw new RuntimeException("出错了...");
                    }
                    System.out.println(value);
                }
            });


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    //累积输入的每个字符串
    public static class MyMapFunction implements MapFunction<String,String>, CheckpointedFunction
    {

        private ListState<String> listState;

        @Override
        public String map(String value) throws Exception {
            listState.add(value);
            return listState.get().toString();
        }

        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("MyMapFunction.snapshotState");

        }

        //初始化状态。  在Task重启时，从之前的备份中帮你恢复状态。 只在Task创建时，执行一次！
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("MyMapFunction.initializeState");

            listState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>("strs", String.class));
        }
    }

}
